package com.atguigu.day02.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;

import java.util.Properties;

public class Flink03_Source_Kafka {
    public static void main(String[] args) {
         Configuration conf = new Configuration();
          //配置web页面固定端口
          conf.setInteger("rest.port", 20000);
          //创建流环境
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
          //设置并行度
          env.setParallelism(2);

        Properties props = new Properties();
        props.put("bootstrap.servers","hadoop162:9092,hadoop163:9092,hadoop164:9092");
        //设置消费者组
        props.put("group.id","Flink03_Source_Kafka");
        //设置从哪开始消费
        //latest:消费的数据如果没有上次记录，则从最新位置开始消费。如果有上次消费的记录，则从上次的位置开始消费
        props.put("auto.offset.reset","latest");
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>("sensor", new SimpleStringSchema(), props));
        stream.print();
        //schema 反序列化, includeMetaDate:是否包含元数据，消费的是json字符串
        DataStreamSource<ObjectNode> sensor = env.addSource(new FlinkKafkaConsumer<>("sensor", new JSONKeyValueDeserializationSchema(true), props));
        sensor.print();

        try {
              env.execute();
          } catch (Exception e) {
              e.printStackTrace();
          }
    }
}
